﻿// StreamInsight example adapters 
// - Microsoft StreamInsight application examples
// (C) Johan Åhlén, 2010. Released under Apache License 2.0 (http://www.apache.org/licenses/)

using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading;
using System.Globalization;
using System.Net;
using System.Net.Sockets;
using System.IO;
using Microsoft.ComplexEventProcessing;
using Microsoft.ComplexEventProcessing.Adapters;

namespace ProIAdapters.Input.SecurityEventLog
{
    /// <summary>
    /// Input adapter for Security Event Log
    /// 
    /// 
    /// 
    /// </summary>
    public class SessionEventInput : TypedIntervalInputAdapter<SessionEventInterval>
    {
        private readonly static IFormatProvider DateFormatProvider = CultureInfo.GetCultureInfo("en-us").DateTimeFormat;
        private const string DateFormatString = "ddd MMM dd HH:mm:ss yyyy";
        private IntervalEvent<SessionEventInterval> pendingEvent;
        private SecurityLogConfig _config;
        const int streamBufferSize =  100000;
        const int dataArraySize = 100000;

        /// <summary>
        /// Constructor
        /// </summary>
        /// <param name="config">Configuration for this adapter</param>
        public SessionEventInput(SecurityLogConfig config)
        {
            _config = config;
        }

        public override void Start()
        {
            ProduceEvents();
        }

        public override void Resume()
        {
            ProduceEvents();
        }

        protected override void Dispose(bool disposing)
        {
            base.Dispose(disposing);
        }

        /// <summary>
        /// Main loop
        /// </summary>
        private void ProduceEvents()
        {
           Socket clientSocket = new Socket(AddressFamily.InterNetwork,
            SocketType.Stream, ProtocolType.Tcp);

            clientSocket.Connect(new IPEndPoint(
            Dns.GetHostAddresses(_config.ServerName)[0], _config.Port));

             
            Console.WriteLine("Client is connected.\n");
            using (Stream
                     netStream = new NetworkStream(clientSocket, true),
                     bufStream =
                           new BufferedStream(netStream, streamBufferSize))
            {
                // Check whether the underlying stream supports seeking.
                Console.WriteLine("NetworkStream {0} seeking.\n",
                    bufStream.CanSeek ? "supports" : "does not support");


                if (bufStream.CanRead)
                {
                    DateTime startTime;
                    double networkTime, bufferedTime = 0;
                    int bytesReceived = 0;
                    byte[] receivedData = new byte[dataArraySize];
                    // Receive data using the NetworkStream.  TEST LATER
                    //Console.WriteLine("Receiving data using NetworkStream.");
                    //startTime = DateTime.Now;
                    //while (bytesReceived < numberOfLoops * receivedData.Length)
                    //{
                    //    bytesReceived += netStream.Read(
                    //        receivedData, 0, receivedData.Length);
                    //}
                    //networkTime = (DateTime.Now - startTime).TotalSeconds;
                    //Console.WriteLine("{0} bytes received in {1} seconds.\n",
                    //    bytesReceived.ToString(),
                    //    networkTime.ToString("F1"));

                    // Receive data using the BufferedStream.
                    Console.WriteLine("Receiving data using BufferedStream.");
                    bytesReceived = 0;
                    startTime = DateTime.Now;

                    int numBytesToRead = receivedData.Length;

                    while (numBytesToRead > 0)
                    {
                        // Read may return anything from 0 to numBytesToRead.
                        int n = bufStream.Read(receivedData, 0, receivedData.Length);
                        // The end of the file is reached.
                        if (n == 0)
                            break;
                        bytesReceived += n;
                        numBytesToRead -= n;
                    }
                    var currEvent = default(IntervalEvent<SessionEventInterval>);
                    object queue = new object();
                    //IEnumerable<SecurityLogEvent> queue = new IEnumerable<SecurityLogEvent>;

                    using (var streamReader = new StreamReader(bufStream))
                    //using (var streamReader = new StreamReader(response.GetResponseStream()))
                    {
                        EnqueueCtiEvent(DateTimeOffset.Now);
                        try
                        {
                            // Loop until stop signal
                            while (AdapterState != AdapterState.Stopping)
                            {
                                if (pendingEvent != null)
                                {
                                    currEvent = pendingEvent;
                                    pendingEvent = null;
                                }
                                else
                                {
                                    try
                                    {
                                        // Read from source
                                        //streamReader.
                                        var line = streamReader.ReadLine();

                                        // Parse
                                        string[] fields = line.Split('|');
                                        var logevent = new SessionEventInterval();
                                        //TODO need to add the rest of the data
                                        logevent.MachineName = fields[6];
                                        logevent.TargetUserSid = fields[7];


                                        // Produce INSERT event
                                        currEvent = CreateInsertEvent();
                                        currEvent.StartTime = DateTimeOffset.Now;

                                        currEvent.Payload = logevent;
                                        pendingEvent = null;
                                        //PrintEvent(currEvent);
                                        Enqueue(ref currEvent);

                                        // Also send a CTI event
                                        EnqueueCtiEvent(DateTimeOffset.Now);

                                    }
                                    catch (Exception ex)
                                    {
                                        // Error handling should go here
                                    }
                                }
                            }

                            if (pendingEvent != null)
                            {
                                currEvent = pendingEvent;
                                pendingEvent = null;
                            }

                            PrepareToStop(currEvent);
                            Stopped();
                        }
                        catch (AdapterException e)
                        {
                            Console.WriteLine(this.GetType().ToString() + ".ProduceEvents - " + e.Message + e.StackTrace);
                        }
                    }
                }

            }
             }


        private string GetLog()
        {
            string result;

            switch (_config.Mode)
            {
                case SecurityLogMode.Sample:
                    //link will be to WCF service
                    result = "http://stream.twitter.com/1/statuses/sample.json";
                    break;
                case SecurityLogMode.Firehose:
                    //link will be to WCF service
                    result = "http://stream.twitter.com/1/statuses/firehose.json";
                    break;
                case SecurityLogMode.Filter:
                    //link will be to WCF service
                    result = "scsproi003:55900";
                    break;
                default:
                    throw new Exception("Invalid Config Mode");
            }

            if (!string.IsNullOrEmpty(_config.Parameters))
                result += "?" + _config.Parameters;

            return result;
        }

  
        private void PrepareToStop(IntervalEvent<SessionEventInterval> currEvent)
        {
            //EnqueueCtiEvent(DateTime.Now);
            if (currEvent != null)
            {
                // Do this to avoid memory leaks
                ReleaseEvent(ref currEvent);
            }
        }

        private void PrepareToResume(IntervalEvent<SessionEventInterval> currEvent)
        {
            pendingEvent = currEvent;
        }

        private string Unquote(string str)
        {
            return str.Trim('"');
        }

        /// <summary>
        /// Debugging function
        /// </summary>
        /// <param name="evt"></param>
        private void PrintEvent(PointEvent<SecurityLogEvent> evt)
        {
        }


    }
}
